package com.lost.octopus.es.processor.mq.consumer;

import com.lost.octopus.es.processor.mq.consumer.capture.DataCaptureManager;
import com.lost.octopus.es.processor.mq.consumer.dto.ChangeDataCaptureInfo;
import lombok.extern.log4j.Log4j2;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.function.Consumer;

/**
 * 数据捕获消费者
 * @author zhangbh
 * @date 2025/4/22 11:36
 */
@Log4j2
@Component("flinkDataConsumer")
public class FlinkDataConsumer implements Consumer<Message<ChangeDataCaptureInfo>> {

    @Resource
    private DataCaptureManager dataCaptureManager;

    @Override
    public void accept(Message<ChangeDataCaptureInfo> message) {
        ChangeDataCaptureInfo payload = message.getPayload();
        boolean isDeal = dataCaptureManager.acceptCapture(payload);
        log.debug("数据 id ={} 捕获处理结果：{}", message.getHeaders().getId(), isDeal);
    }
}
